package cn.enjoyedu.rejectmsg;

import cn.enjoyedu.exchange.direct.DirectProducer;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * channel.basicReject(envelope.getDeliveryTag(), false);
 * channel.basicReject(envelope.getDeliveryTag(), true);
 * 第二个参数如果是true就会允许重新投递
 * 为false表示拒绝且不需要重新投递，相当于消费了但是消费者处理失败，进入死信队列
 * 消息过期也会被投递到死信队列
 * 消息不配置的话默认是不过期的
 * 死信交换器DLX，默认的
 */
public class RejectRequeuConsumer {

    public static void main(String[] argv)
            throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");

        // 打开连接和创建频道，与发送端一样
        Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();
        channel.exchangeDeclare(DirectProducer.EXCHANGE_NAME,
                "direct");

        /*声明一个队列*/
        String queueName = "focuserror";
        channel.queueDeclare(queueName, false, false,
                false, null);

        /*绑定，将队列和交换器通过路由键进行绑定*/
        String routekey = "error";/*表示只关注error级别的日志消息*/
        channel.queueBind(queueName, DirectProducer.EXCHANGE_NAME, routekey);

        System.out.println("waiting for message........");

        /*声明了一个消费者*/
        final Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                try {
                    String message = new String(body, "UTF-8");
                    System.out.println("Received["
                            + envelope.getRoutingKey()
                            + "]" + message);
                    throw new RuntimeException("处理异常" + message);
                } catch (Exception e) {
                    e.printStackTrace();
                    channel.basicReject(envelope.getDeliveryTag(), false);
                }

            }
        };
        /*消费者正式开始在指定队列上消费消息*/
        channel.basicConsume(queueName, false, consumer);


    }

}
